package com.xunk.equipment.websocket;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.xunk.auth.api.RemoteAuthService;
import com.xunk.common.core.constant.HttpStatus;
import com.xunk.common.core.utils.DataUtils;
import com.xunk.common.core.utils.SpringUtils;
import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;

import org.springframework.stereotype.Component;

import java.io.IOException;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.*;

@ServerEndpoint(value = "/ws/{sid}")
@Component
public class WebSocketServer {
    private final static Logger log = LoggerFactory.getLogger(WebSocketServer.class);
    //与某个客户端的连接会话，需要通过它来给客户端发送数据
    private Session session;
    //接收用户的sid，指定需要推送的用户
    private String sid;

    private static final String TOKEN = "token";
    private static final String START_TIME = "Start Time";
    private static final String PONG = "PONG";
    private static final Long INTERVAL = 3L;


    /**
     * 连接成功后调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("sid") String sid) {
        this.session = session;
        session.getUserProperties().put(START_TIME, System.currentTimeMillis());
        String queryString = session.getQueryString();
        Boolean authentication = authentication(queryString, sid);
        if (!authentication) {
            log.info("链接不合法，断开 {}", sid);
            String s = buildMessage(HttpStatus.ERROR, "鉴权失败，断开链接。", 0, null);
            sendMessage(s);
            onClose();
            return;
        }
        List<WebSocketServer> servers = WebsocketManager.WEBSOCKET_MAP.get(sid);
        if (CollectionUtils.isNotEmpty(servers)) {
//            if (servers.size() >= 10) {
//                log.info("客户端个数已经超出限制 {}", sid);
//                close(session);
//                return;
//            }
            String s = buildMessage(HttpStatus.SUCCESS, "警告：账号正在其他客户端登录。", 0, null);
            servers.forEach(server -> server.sendMessage(s));
        } else {
            servers = new ArrayList<>();
        }
//        if (Objects.nonNull(webSocketServer)) {
//            webSocketServer.sendMessage("在其他地方登录，当前断开链接");
//            close(webSocketServer.session);
//        }
        WebsocketManager.addOnlineCount();           //在线数加1
        // 连接成功后默认Pong一次
        setPongTime();
        servers.add(this);
        WebsocketManager.WEBSOCKET_MAP.put(sid, servers); //加入map中
        log.info("有新窗口开始监听:" + sid + ",当前在线人数为" + WebsocketManager.getOnlineCount());
        this.sid = sid;
        String s = buildMessage(HttpStatus.SUCCESS, "连接成功。", 0, null);
        sendMessage(s);
    }

    private Boolean authentication(String queryString, String sid) {
        if (DataUtils.isEmpty(queryString)) {
            return false;
        }
        String[] split = queryString.split("&");
        Map<String, String> paramMap = new HashMap<>();
        for (String s : split) {
            String[] params = s.split("=");
            if (params.length != 2) {
                continue;
            }
            paramMap.put(params[0], params[1]);
        }
        String token = paramMap.get(TOKEN);
        if (DataUtils.isEmpty(token)) {
            log.info("参数异常");
            return false;
        }
        RemoteAuthService remoteAuthService = SpringUtils.getBean(RemoteAuthService.class);
        String s = remoteAuthService.checkToken(token);
        if (DataUtils.isEmpty(s)) {
            log.info("鉴权失败");
            return false;
        }
        JSONObject result = JSON.parseObject(s);
        String userId = result.getString("user_id");
        if (Objects.isNull(userId)) {
            log.info("鉴权失败 message = {} ", result.getString("msg"));
            return false;
        }
        Boolean active = result.getBoolean("active");
        if (!active) {
            log.info("账号已停用");
            return false;
        }

        Long exp = result.getLong("exp");
        return Objects.equals(userId, sid) && exp > 0;
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        // 移除server
        remove();
    }

    private synchronized void remove() {
        if (Objects.nonNull(this.sid)) {
            List<WebSocketServer> servers = WebsocketManager.WEBSOCKET_MAP.get(this.sid);
            Optional.ofNullable(servers).ifPresent(s -> {
                boolean b = s.removeIf(next -> next.session == this.session);
                if (b) {
                    int count = WebsocketManager.subOnlineCount();//在线数减1
                    // 关闭session
                    close(this.session);
                    log.info("有一连接关闭！当前在线人数为 {}", count);
                }
            });
            if (CollectionUtils.isEmpty(servers)) {
                // 移除，垃圾回收
                WebsocketManager.WEBSOCKET_MAP.remove(this.sid);
            }
        }
    }

    private void close(Session session) {
        try {
            log.info("session closed after {} s", getConnectionSeconds());
            if (DataUtils.isNotEmpty(session) && session.isOpen()) {
                session.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息
     */
    @OnMessage
    public void onMessage(String message) {
        log.info("收到来自窗口" + sid + "的信息:" + message);
        if (Objects.equals(message, PONG)) {
            setPongTime();
        }
    }

    /**
     * 发生错误时的回调函数
     *
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("发生错误", error);
    }

    /**
     * 实现服务器主动推送消息
     */
    public void sendMessage(String message) {
        try {
            this.session.getBasicRemote().sendText(message);
        } catch (IOException e) {
            e.printStackTrace();
            log.error("websocket IO异常", e);
        }
    }

    int getConnectionSeconds() {
        long millis = System.currentTimeMillis() - ((Long) this.session.getUserProperties().get(START_TIME));
        return (int) millis / 1000;
    }

    private void setPongTime() {
        session.getUserProperties().put(PONG, LocalDateTime.now());
    }

    public LocalDateTime getPongTime() {
        return (LocalDateTime) session.getUserProperties().get(PONG);
    }

    public String getSessionId() {
        return session.getId();
    }

    public void cleanExpireServer() {
        LocalDateTime now = LocalDateTime.now();
        LocalDateTime pongTime = (LocalDateTime) session.getUserProperties().get(PONG);
        // 如果客户端每2分钟pong一次，那么这里检测差值不能超过2分钟
        long minutes = Duration.between(pongTime, now).toMinutes();
        if (minutes > INTERVAL) {
            String s = buildMessage(HttpStatus.SUCCESS, "超时pong，断开链接。", 0, null);
            sendMessage(s);
            onClose();
        }
    }

    private String buildMessage(Integer code, String msg, Integer type, String data) {
        Map<String, Object> m = new HashMap<>();
        m.put("code", code);
        m.put("msg", msg);
        m.put("type", type);
        m.put("data", data);
        return JSON.toJSONString(m);
    }
}
